package com.spark.mooc.ch7_sparkstreaming.part03_kafkaDataSource

import java.util
import java.util.Properties
import org.apache.kafka.clients.producer.Callback
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.clients.producer.ProducerConfig
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.clients.producer.RecordMetadata

import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import scala.collection.mutable
/**
 * @description:
 * @time: 2020/11/29 16:57
 * @author: lhy
 */
object KafkaWordProducer {
    def main(args: Array[String]): Unit = {
        if (args.length < 4) {
            /*参数说明：
            <metadataBrokerList> :Broker的地址
            <topic>
            <messagesPerSec> ：每秒钟想发几条消息给topic
            <wordsPerMessage>：每条消息要包含几个单词
             */
            System.err.println("Usage: KafkaWordCountProducer <metadataBrokerList> <topic> <messagesPerSec> <wordsPerMessage>")
            System.exit(1)
        }
        val Array(brokers, topic, messagesPerSec, wordsPerMessage) = args       //提取器
        // Zookeeper connection properties
        val props = new util.HashMap[String,Object]()
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokers)
        // key value 可序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")

        val producer = new KafkaProducer[String,String](props)
        // Send some messages
        while (true){
            (1 to messagesPerSec.toInt).foreach{ messageNum =>
                val str: String = (1 to wordsPerMessage.toInt)
                  .map(x => scala.util.Random.nextInt(10).toString)
                  .mkString(" ")
                print(str)
                println()
                val message = new ProducerRecord[String,String](topic,null,str)
                producer.send(message)
            }
            Thread.sleep(1000)
        }
    }
}
